package cn.z3inc.kafkademo.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Future;

/**
 *  生产者测试类
 *                 网络传数据 ，数据需要做序列化
 * @author z3inc
 * @since 2024/11/17
 */
public class ProducerTest {
    public static void main(String[] args) {
        //  0.组装生产者对象需要的配置参数
        Map<String, Object> configMap = new HashMap<>();
        configMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // 对生产的数据k和v，进行序列化操作
        configMap.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//key使用 StringSerializer 做序列化
        configMap.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//value使用 StringSerializer 做序列化

        //  1.创建生产者对象KafkaProducer, 需要设定泛型:  数据的类型约束
        KafkaProducer<String,String> producer = new KafkaProducer<String,String>(configMap);

        //  2.创建数据对象:ProducerRecord
        //          构建数据时，需要传递三个参数
        //          第一个参数: 指定主题名称， 如果kafka中不存在则会自动创建
        //          第二个参数: 数据的key
        //          第三个参数: 数据的value（真正的数据）
        ProducerRecord<String, String> record = new ProducerRecord<String, String>("topic","key", "hello kafka");

        //   3.通过生产者对象将数据发送给kafka
        Future<RecordMetadata> send = producer.send(record);
        try {
            RecordMetadata recordMetadata = send.get();
            System.out.println(recordMetadata);
        } catch (Exception e) {
            e.printStackTrace();
        }

        //   4.关闭生产者对象
        producer.close();
    }
}
